package com.c6906.vdam.dcs.pool;

import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.producer.Producer;

public class ProducerFactory extends BasePooledObjectFactory<Producer> {
    private final ClientServiceProvider provider;
    private final ClientConfiguration configuration;
    private final String topic;

    public ProducerFactory(String endpoint, String topic) throws ClientException {
        this.provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().enableSsl(true).setEndpoints(endpoint);
        this.configuration = builder.build();
        this.topic = topic;
    }

    @Override
    public Producer create() throws Exception {
        return provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();
    }

    @Override
    public PooledObject<Producer> wrap(Producer producer) {
        return new DefaultPooledObject<>(producer);
    }

    @Override
    public void destroyObject(PooledObject<Producer> p) throws Exception {
        p.getObject().close();
    }
}
